package com.linkstec.redis;

import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.alibaba.fastjson.JSON;

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;

/**
 * Redis集群工具类
 *
 * @author
 */
public class JedisClusterUtil {

    private static final Logger logger = LogManager.getLogger();

    public static JedisPool pool = null;

    private static String redisHostsAndPortsCluster;

    private static String keyNameSpace;
    private static String eventFlowKeyNameSpace;

    private static JedisCluster jedisCluster = null;
    private static JedisPoolConfig poolConfig = null;
    private static String passWord = null;

    @SuppressWarnings("rawtypes")
    public static void setConfig(Map cfg) {
        redisHostsAndPortsCluster = (String) cfg.get("redisHostsAndPortsCluster");
        keyNameSpace = (String) cfg.get("keyNameSpace");
        eventFlowKeyNameSpace = (String) cfg.get("eventFlowKeyNameSpace");
        passWord = (String) cfg.get("redisPassWord");
        logger.info("keyNameSpace:" + keyNameSpace);
        logger.info("eventFlowKeyNameSpace:" + eventFlowKeyNameSpace);
        logger.info("redisHostsAndPortsCluster:" + redisHostsAndPortsCluster);
        logger.info("passWord:" + passWord);
    }

    //
    public static JedisCluster getJedisCluster() {
        if (jedisCluster == null) {
            synchronized (JedisClusterUtil.class) {
                if (jedisCluster == null) {
                    poolConfig = new JedisPoolConfig();
                    poolConfig.setMaxTotal(256);
                    poolConfig.setMaxIdle(10);
                    poolConfig.setMinIdle(10);
                    poolConfig.setMaxWaitMillis(3000L);
                    poolConfig.setTestOnReturn(false);
                    poolConfig.setTestOnBorrow(false);
                    poolConfig.setTestWhileIdle(true);
                    poolConfig.setTimeBetweenEvictionRunsMillis(30000);
                    poolConfig.setNumTestsPerEvictionRun(3);

                    Set<HostAndPort> nodes = new HashSet<HostAndPort>();
                    String[] hostAndPortArray = redisHostsAndPortsCluster.split(",");
                    for (String hostAndPort : hostAndPortArray) {
                        String host = hostAndPort.split(":")[0];
                        int port = Integer.valueOf(hostAndPort.split(":")[1]);
                        nodes.add(new HostAndPort(host, port));
                    }
                    jedisCluster = new JedisCluster(nodes, 5000, 999, 10, passWord, poolConfig);
                }
            }
        }

        return jedisCluster;

    }

    // set key-value
    public static void set(String key, Object value) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            jedisCluster.set(redisKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    // set key-value 增强set
    public static String set(final String key, final String value, final String nxxx, final String expx,
            final long time) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        String result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = value;
                }
            }
            result = jedisCluster.set(redisKey, strVal, nxxx, expx, time);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    // set key-value and set key expire time
    public static void setExpire(String key, int expireTime, Object value) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            jedisCluster.setex(redisKey, expireTime, strVal);
        } catch (Exception e) {
            logger.error("set key " + redisKey + "error:", e);
        }
    }

    // get value by key
    @SuppressWarnings("unchecked")
    public static <T> T get(String key, Class<T> clazz) {
        String redisKey = keyNameSpace + key;
        T value = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            String strVal = jedisCluster.get(redisKey);
            if (strVal != null && !String.class.equals(clazz)) {
                value = JSON.parseObject(strVal, clazz);
            } else {
                value = (T) strVal;
            }
        } catch (Exception e) {
            logger.error("", e);
        }
        return value;
    }

    // put a hashmap to redis
    public static void putHashMap(String key, Map<String, String> map) throws Exception {
        String redisKey = keyNameSpace + key;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            jedisCluster.hmset(redisKey, map);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    // get hash map element
    public static String hget(String key, String hashKey) throws Exception {
        String redisKey = keyNameSpace + key;
        String result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.hget(redisKey, hashKey);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    // get hash map element
    public static Map<String, String> getMap(String key) {
        String redisKey = keyNameSpace + key;
        Map<String, String> result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.hgetAll(redisKey);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public static void hSet(String key, String hashKey, Object hashValue) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (hashValue != null) {
                if (!(hashValue instanceof String)) {
                    strVal = JSON.toJSONString(hashValue);
                } else {
                    strVal = (String) hashValue;
                }
            }
            jedisCluster.hset(redisKey, hashKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public static boolean hContains(String key, String hashKey) {
        String redisKey = keyNameSpace + key;
        boolean result = false;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.hexists(redisKey, hashKey);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static Set<String> hKeys(String key) {
        String redisKey = keyNameSpace + key;
        Set<String> result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.hkeys(redisKey);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static void hDel(String key, String hashKey) {
        String redisKey = keyNameSpace + key;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            jedisCluster.hdel(redisKey, hashKey);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public static Map<String, String> hScan(String key) {
        String redisKey = keyNameSpace + key;
        Map<String, String> result = new HashMap<String, String>();
        ScanParams params = new ScanParams();
        params.count(2000);
        try {
            JedisCluster jedisCluster = getJedisCluster();
            String cursor = "0";
            while (true) {
                ScanResult<Entry<String, String>> scanResult = jedisCluster.hscan(redisKey, cursor, params);
                for (Entry<String, String> entry : scanResult.getResult()) {
                    result.put(entry.getKey(), entry.getValue());
                }
                cursor = scanResult.getStringCursor();
                if (cursor.equals("0")) {
                    break;
                }
            }

        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static void del(String key) {
        String redisKey = keyNameSpace + key;
        try {
            if (!StringUtils.isEmpty(key)) {
                JedisCluster jedisCluster = getJedisCluster();
                jedisCluster.del(redisKey);
            }
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public static void del(Collection<String> keys) {
        try {
            if (null != keys && !keys.isEmpty()) {
                JedisCluster jedisCluster = getJedisCluster();
                for (String key : keys) {
                    String redisKey = keyNameSpace + key;
                    jedisCluster.del(redisKey);
                }
            }
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public static void delByPrefix(String keyPrefix) {
        String redisKey = keyNameSpace + keyPrefix;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            Map<String, JedisPool> jedisPoolMap = jedisCluster.getClusterNodes();
            for (JedisPool jedisPool : jedisPoolMap.values()) {
                Jedis jedis = jedisPool.getResource();
                Set<String> keySet = jedis.keys(redisKey);
                if (null == keySet || keySet.isEmpty()) {
                    for (String key : keySet) {
                        jedisCluster.del(key);
                    }
                    jedis.close();
                }
            }
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    @Deprecated
    public static void addSet(String key, Object value) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            jedisCluster.sadd(redisKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        } finally {
//			returnResource(jedisCluster);
        }
    }

    public static Long sadd(String key, Object value) {
        Long result = null;
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            result = jedisCluster.sadd(redisKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        } finally {
//			returnResource(jedisCluster);
        }
        return result;
    }

    public static void srem(String key, Object value) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            jedisCluster.srem(redisKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        } finally {
//			returnResource(jedisCluster);
        }
    }

    public static boolean setContains(String key, Object value) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        boolean result = false;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            result = jedisCluster.sismember(redisKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static Set<String> sMembers(String key) {
        String redisKey = keyNameSpace + key;
        Set<String> result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.smembers(redisKey);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static String srandmember(String key) {
        String redisKey = keyNameSpace + key;
        String result = null;
        JedisCluster jedisCluster = getJedisCluster();
        try {
            result = jedisCluster.srandmember(redisKey);
        } catch (Exception e) {
            logger.error("", e);
            throw e;
        }

        return result;
    }

    public static Set<String> sScan(String key) {
        String redisKey = keyNameSpace + key;
        Set<String> result = new HashSet<String>();
        ScanParams params = new ScanParams();
        params.count(2000);
        JedisCluster jedisCluster = getJedisCluster();
        String cursor = "0";
        while (true) {
            try {
                ScanResult<String> scanResult = jedisCluster.sscan(redisKey, cursor, params);
                result.addAll(scanResult.getResult());
                cursor = scanResult.getStringCursor();
                if (cursor.equals("0")) {
                    break;
                }
            } catch (Exception e) {
                logger.error("", e);
                cursor = "0";
                result.clear();
            }
        }
        return result;
    }

    public static Long incrBy(final String key, final long integer) {
        String redisKey = keyNameSpace + key;
        Long result = 0L;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.incrBy(redisKey, integer);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public static Long incrByBatch(final String key, final long increment) {
        String redisKey = eventFlowKeyNameSpace + key;
        Long result = 0L;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.incrBy(redisKey, increment);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public static void addList(String key, Object value) {
        String redisKey = keyNameSpace + key;
        String strVal = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            if (value != null) {
                if (!(value instanceof String)) {
                    strVal = JSON.toJSONString(value);
                } else {
                    strVal = (String) value;
                }
            }
            jedisCluster.rpush(redisKey, strVal);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public static List<String> lrange(String key) {
        String redisKey = keyNameSpace + key;
        List<String> result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.lrange(redisKey, 0, -1);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static Long lpush(String key, final String... value) {
        String redisKey = eventFlowKeyNameSpace + key;
        Long result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.lpush(redisKey, value);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static Long rpush(String key, final String... value) {
        String redisKey = eventFlowKeyNameSpace + key;
        Long result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.rpush(redisKey, value);
        } catch (Exception e) {
            logger.error("", e);
        }

        return result;
    }

    public static void expired(String key, int seconds) {
        String redisKey = keyNameSpace + key;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            jedisCluster.expire(redisKey, seconds);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public static boolean exists(String key) {
        String redisKey = keyNameSpace + key;
        boolean result = false;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.exists(redisKey);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public static ScanResult<String> scan(final String cursor, ScanParams scanParams) {
        ScanResult<String> result = null;
        try {
            JedisCluster jedisCluster = getJedisCluster();
            result = jedisCluster.scan(cursor, scanParams);
        } catch (Exception e) {
            logger.error("", e);
        }
        return result;
    }

    public static void clusterNodesScan(String cursor, ScanParams scanParams) {

        AtomicInteger count = new AtomicInteger();
        JedisCluster cluster = getJedisCluster();
        Map<String, JedisPool> clusterNodes = cluster.getClusterNodes();
        for (Iterator<JedisPool> it = clusterNodes.values().iterator(); it.hasNext();) {
            JedisPool pool = it.next();
            Jedis jedis = null;
            try {
                jedis = pool.getResource();

                do {
                    ScanResult<String> result = jedis.scan(cursor, scanParams);
                    cursor = result.getStringCursor();
                    for (String key : result.getResult()) {
                        count.getAndIncrement();
                        // 做自己的业务逻辑
                        logger.info(key);
                    }
                } while (!cursor.equals("0"));
                logger.info("共从Redis中获取到：{}条数据", count);
                logger.info("获取Redis数据完成");
            } catch (Exception e) {
                logger.error("RedisClusterCacheUtil.keys occurred Exceotion", e);
            } finally {
                if (null != jedis)
                    jedis.close();
            }
        }
    }

}